package org.halk;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * @author halk
 * @date 2023/3/9
 */
public class ZookeeperDemo {

    private CuratorFramework client;

    @Before
    public void createClient() {
//      CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", 60000, 5000, new RetryNTimes(5, 5000));
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryPolicy);
        client.start();
    }

    /**
     * 添加永久节点
     *
     * @throws Exception
     */
    @Test
    public void createPath() throws Exception {
        String s = client.create().forPath("/my/mypath", "abc".getBytes(StandardCharsets.UTF_8));
        System.out.println(s);
    }

    /**
     * 添加临时节点
     *
     * @throws Exception
     */
    @Test
    public void createTmpPath() throws Exception {
        String s = client.create().withMode(CreateMode.EPHEMERAL).forPath("/my/mypath/tmp", "abc".getBytes(StandardCharsets.UTF_8));
        System.out.println(s);

        System.in.read();
    }

    /**
     * 获取节点下的数据
     *
     * @throws Exception
     */
    @Test
    public void getData() throws Exception {
        byte[] bytes = client.getData().forPath("/my/mypath");
        System.out.println(new String(bytes));
    }

    /**
     * 设置值
     *
     * @throws Exception
     */
    @Test
    public void setData() throws Exception {
        Stat stat = client.setData().forPath("/my/mypath", "18255334455".getBytes(StandardCharsets.UTF_8));
        System.out.println(stat);

        byte[] bytes = client.getData().forPath("/my/mypath");
        System.out.println(new String(bytes));
    }

    /**
     * 如果父级不存在就创建
     *
     * @throws Exception
     */
    @Test
    public void createParentNode() throws Exception {
        String s = client.create().creatingParentsIfNeeded().forPath("/my/needPath/yy", "need path".getBytes(StandardCharsets.UTF_8));
        System.out.println(s);
    }

    /**
     * 删除
     *
     * @throws Exception
     */
    @Test
    public void deletePath() throws Exception {
        client.delete().deletingChildrenIfNeeded().forPath("/my");
    }

    /**
     * 监听节点对应的值改变
     *
     * @throws Exception
     */
    @Test
    public void watchTest() throws Exception {
        NodeCache nodeCache = new NodeCache(client, "/test2");
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("node change");
                byte[] bytes = client.getData().forPath("/test2");
                System.out.println(new String(bytes));
            }
        });

        nodeCache.start();

        System.in.read();
    }

    /**
     * 读锁的的前面必须是读锁，或者没有锁
     * 尝试获取读锁
     *
     * @throws Exception
     */
    @Test
    public void testGetReadLock() throws Exception {
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/lock1");
        //获取读锁
        InterProcessMutex interProcessMutex = readWriteLock.readLock();
        System.out.println("开始尝试获取锁！");
        //一直阻塞，直到获取到锁
        interProcessMutex.acquire();

        System.out.println("拿到读锁了");

        for (int i = 0; i < 15; i++) {
            Thread.sleep(1000);
            System.out.println(i);
        }
        interProcessMutex.release();
    }

    /**
     * 尝试获取读锁2
     *
     * @throws Exception
     */
    @Test
    public void testGetReadLock2() throws Exception {
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/lock1");
        //获取读锁
        InterProcessMutex interProcessMutex = readWriteLock.readLock();
        System.out.println("开始尝试获取锁！");
        //一直阻塞，直到获取到锁
        interProcessMutex.acquire();

        System.out.println("拿到读锁了");

        for (int i = 0; i < 15; i++) {
            Thread.sleep(1000);
            System.out.println(i);
        }
        interProcessMutex.release();
    }

    /**
     * 写锁前面不能有任何锁
     * 尝试获取写锁
     *
     * @throws Exception
     */
    @Test
    public void testGetWriteLock() throws Exception {
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/lock1");
        //获取读锁
        InterProcessMutex interProcessMutex = readWriteLock.writeLock();
        System.out.println("开始尝试获取锁！");
        //一直阻塞，直到获取到锁
        interProcessMutex.acquire();

        System.out.println("拿到写锁了");

        for (int i = 0; i < 15; i++) {
            Thread.sleep(1000);
            System.out.println(i);
        }
        interProcessMutex.release();
    }

    /**
     * 尝试获取写锁2
     *
     * @throws Exception
     */
    @Test
    public void testGetWriteLock2() throws Exception {
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/lock1");
        //获取读锁
        InterProcessMutex interProcessMutex = readWriteLock.writeLock();
        System.out.println("开始尝试获取锁！");
        //一直阻塞，直到获取到锁
        interProcessMutex.acquire();

        System.out.println("拿到写锁了");

        for (int i = 0; i < 15; i++) {
            Thread.sleep(1000);
            System.out.println(i);
        }
        interProcessMutex.release();
    }

    @Test
    public void clientDemo() throws Exception {
        InterProcessMutex lock = new InterProcessMutex(client, "/my");
        if (lock.acquire(2, TimeUnit.SECONDS)) {
            try {
                // do some work inside of the critical section here
                System.out.println("拿到锁，休眠10S");
                Thread.sleep(10000);
                System.out.println("休眠结束");
            } finally {
                lock.release();
            }
        }
    }

    @Test
    public void test() throws Exception {
        InterProcessMutex lock = new InterProcessMutex(client, "/my");
        if (lock.acquire(2, TimeUnit.SECONDS)) {
            try {
                // do some work inside of the critical section here
                System.out.println("拿到锁");
            } finally {
                lock.release();
            }
        } else {
            System.out.println("未拿到锁");
        }
    }
}
